/*
 * @Author: lokei
 * @Date: 2022-08-28 22:46:40
 * @LastEditors: lokei
 * @LastEditTime: 2022-08-30 08:31:48
 * @Description: 
 */
package cn.lokei.gateway.handler.message;

import cn.lokei.gateway.service.ChannelSupervise;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.channel.group.ChannelGroup;

public class ReceiveHandler {

	private ChannelSupervise channelSupervise = ChannelSupervise.getBean(ChannelSupervise.class);

	public void receiveMessage(MqttMessage mqttMessage) {
		MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
		String topic = mqttPublishMessage.variableHeader().topicName();
		ChannelGroup channelGroup = channelSupervise.subChannels(topic);
		if ((!(channelGroup == null)) && channelGroup.size() > 0) {
			MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false,
					mqttPublishMessage.fixedHeader().qosLevel(), false,
					mqttPublishMessage.fixedHeader().remainingLength());
			// 构建返回报文， 可变报头
			MqttPublishVariableHeader mqttMessageIdVariableHeader = new MqttPublishVariableHeader(topic,
					mqttPublishMessage.variableHeader().packetId());
			MqttPublishMessage mqttMessagePublish = new MqttPublishMessage(mqttFixedHeader, mqttMessageIdVariableHeader,
					mqttPublishMessage.payload());
			channelGroup.writeAndFlush(mqttMessagePublish);
		}
	}
}
